/*
 * Copyright 2011 es-planning Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package jp.arctos.ursus.extension.net;

import java.io.IOException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.locks.ReentrantLock;

/**
 * SessionWriter base class.
 * sends data (writing to channel)
 *
 */
abstract class AbstractSessionWriter implements SessionWriter {

    /** Send request holding queue. */
    ArrayBlockingQueue<SendRequest> queue;

    /** Lock for access to queue. */
    ReentrantLock lock = new ReentrantLock();
    
    /** flag indicates whether requests are accepted. */
    boolean isStopWrite = false;

    /**
     * {@inheritDoc}
     * 
     */
    public void initQueue(int queueSize) {
        queue = new ArrayBlockingQueue<SendRequest>(queueSize);
    }

    /**
     * {@inheritDoc}
     * 
     */
    public Queue<SendRequest> getQueue() {
        return queue;
    }

    /**
     * {@inheritDoc}
     * 
     */
    public void stopWrite() {
        isStopWrite = true;
    }

    /**
     * {@inheritDoc}
     * 
     */
    public boolean write(SelectorHandler handler, SelectableChannel channel, SendRequest request) {
        if (isStopWrite || queue == null) return false;

        boolean result = false;
        lock.lock();
        try {
            result = queue.offer(request);
            if (result && queue.size() == 1) {
                handler.addInterestOps(channel, SelectionKey.OP_WRITE);
                handler.wakeup();
            }
        } finally {
            lock.unlock();
        }

        return result;
    }

    /**
     * {@inheritDoc}
     * <p>
     * Writes data at the head of the send request holding queue to the channel
     * If all data is not read, the rest will be sent on the next call.
     * </p>
     */
    public SendRequest onWritable(SelectorHandler handler, SelectableChannel channel) throws Exception {
        SendRequest completed = null;
        lock.lock();
        try {
            SendRequest request = null;
            request = queue.peek();
            if (request != null) {
                doWrite(channel, request);
                if (!request.getBuffer().hasRemaining()) {
                    completed = queue.poll();
                    if (queue.isEmpty()) {
                        handler.removeInterestOps(channel, SelectionKey.OP_WRITE);
                    }
                }
            }
        } finally {
            lock.unlock();
        }
        return completed;
    }

    /**
     * Write send request data to the channel.
     * 
     * @param channel Channel
     * @param request Send request data
     * @throws IOException
     */
    abstract void doWrite(SelectableChannel channel, SendRequest request) throws IOException;
}
